package com.paradox.aggregation.provider.controller;

import cn.hutool.core.lang.UUID;
import cn.hutool.json.JSONUtil;
import com.paradox.aggregation.provider.config.RabbitMqConfig;
import com.paradox.aggregation.provider.config.dataSource.DataSource;
import com.paradox.aggregation.provider.config.dataSource.DataSourceSwitcher;
import com.paradox.aggregation.provider.entity.Providers;
import com.paradox.aggregation.provider.lock.LockService;
import com.paradox.aggregation.provider.mapper.ProvidersMapper;
import com.paradox.aggregation.providerAPI.api.RpcTestService;
import com.paradox.aggregation.providerAPI.config.MessageVo;
import com.paradox.rpc.proxy.ProxyFactory;
import io.seata.core.context.RootContext;
import io.seata.spring.annotation.GlobalTransactional;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.CharSet;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.checkerframework.checker.units.qual.K;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpHeaders;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.servlet.http.HttpServletRequest;
import java.nio.charset.Charset;
import java.util.Date;
import java.util.Enumeration;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutionException;

@RestController
@RequestMapping("/provider")
@Slf4j
public class ProviderController {

    //@Autowired
    //private MongoTemplate mongoTemplate;
    @Autowired
    private ProvidersMapper providersMapper;
    @Autowired
    private LockService lockService;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @GetMapping("/getSource")
    public String getSource(HttpServletRequest request,@RequestHeader HttpHeaders headers) throws InterruptedException {
        String token=request.getHeader("tokenInfo");
        String token2=headers.getFirst("tokenInfo");
        Enumeration<String> enumeration=request.getAttributeNames();
        //Thread.sleep(1200);
        return token+token2;
    }

    public void testMongoDB(){
        //mongoTemplate.findAll(xxx.class)
    }

    @Transactional
    @GetMapping("/getProviders")
    public Boolean testMysql(){
        String xid= RootContext.getXID();
        System.out.println("xid:"+xid);
        providersMapper.insert(new Providers("p002","fuwu",new Date()));
        int x=1/0;
        providersMapper.insert(new Providers("p003","fuwu",new Date()));

        return true;
    }

    @GetMapping("/getProviders1")
    @DataSource(DataSourceSwitcher.READ)
    public List<Providers> getProviders1(){
        //DynamicDatasourceAutoConfiguration.setLookupKey(DynamicDatasourceAutoConfiguration.READ);
        return providersMapper.selectList(null);
    }

    @GetMapping("/getProviders2")
    @DataSource(DataSourceSwitcher.WRITE)
    public List<Providers> getProviders2(){
        //DynamicDatasourceAutoConfiguration.setLookupKey(DynamicDatasourceAutoConfiguration.WRITE);
        return providersMapper.selectList(null);
    }

    /**
     * test for distributed lock
     * @return
     */
    @GetMapping("/rush")
    public String rush(){
        return lockService.manulRush();
    }

    @GetMapping("/sendRabbitMessage")
    public void sendRabbitMessage(String message){
        CorrelationData correlationData=new CorrelationData(UUID.fastUUID().toString());
        MessageVo vo=new MessageVo(1,message);
        rabbitTemplate.convertAndSend("topicExchange","routing.1",vo,correlationData);
        log.info("send message : {}",vo.toString());
    }

    @GetMapping("/sendRocketMessage")
    public void sendRocketMessage(String message){
        SendResult res=rocketMQTemplate.syncSend("provider",new MessageVo(1,"rocketMessage"));
        log.info(res.toString());
    }

    @GetMapping("/sendRocketMessageAsync")
    public void sendRocketMessageAsync(String message){
        rocketMQTemplate.asyncSend("provider", new MessageVo(2, "rocketMessageAsync"), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info(sendResult.toString());
            }

            @Override
            public void onException(Throwable e) {
                log.error(e.toString());
            }
        });
    }

    @GetMapping("/sendMessageInTransaction")
    public void sendMessageInTransaction(String message){
        Message<String> msg= MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.TAGS,"myTag")
                .setHeader(RocketMQHeaders.TRANSACTION_ID, UUID.fastUUID().toString()).build();
        SendResult res=rocketMQTemplate.sendMessageInTransaction("provider",msg,null);
        log.info(res.toString());
    }

    @GetMapping("/sendKafkaMessage")
    public void sendKafkaMessage(String message) throws ExecutionException, InterruptedException {
        MessageVo vo=new MessageVo(1,message);
        ListenableFuture<org.springframework.kafka.support.SendResult<String, String>> future=kafkaTemplate.send("test", JSONUtil.toJsonStr(vo));
        log.info("send message : {}",vo.toString());
        log.info(future.get().toString());
    }

    @GetMapping("/testRpc")
    public String testRpc(String arg){
        RpcTestService rpcTestService= (RpcTestService) new ProxyFactory().createProxy(RpcTestService.class);
        return rpcTestService.getSource(arg);
    }

    @GlobalTransactional
    @GetMapping("/testSeata")
    public Boolean testSeata(){
        String xid= RootContext.getXID();
        System.out.println("xid:"+xid);
        providersMapper.insert(new Providers("p002","fuwu",new Date()));
        DataSourceSwitcher.setLookupKey(DataSourceSwitcher.WRITE);
        providersMapper.insert(new Providers("p003","fuwu",new Date()));
        int x=1/0;
        return true;
    }
}
